package com.javafactroy.kunlong.batch;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.JobRegistry;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.support.JobRegistryBeanPostProcessor;
import org.springframework.batch.core.converter.DefaultJobParametersConverter;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.JobOperator;
import org.springframework.batch.core.launch.support.SimpleJobOperator;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.*;
import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
import org.springframework.batch.item.database.builder.JdbcPagingItemReaderBuilder;
import org.springframework.batch.item.database.support.SqlPagingQueryProviderFactoryBean;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder;
import org.springframework.batch.item.support.CompositeItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.FileSystemResource;
import org.springframework.stereotype.Component;

import javax.sql.DataSource;
import java.io.Serializable;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;


/*
*
* @author:leijmdas
* @filetype: flatFile, xml, db
* @step:  flow, step
* @date:2023-11-04
*
* */
@Configuration
@EnableBatchProcessing
@Slf4j
public class FileJobConfig implements ApplicationContextAware {

    AtomicInteger batchNo = new AtomicInteger(0);
    private ApplicationContext applicationContext;

    @Autowired
    @Qualifier("mysqlDataSource")
    DataSource pfDatasource;

    @Autowired
    private JobBuilderFactory jobBuilderFactory;


    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public JobRegistryBeanPostProcessor jobRegistrar() throws Exception {
        JobRegistryBeanPostProcessor postProcessor = new JobRegistryBeanPostProcessor();
        postProcessor.setJobRegistry(jobRegistry);
        postProcessor.setBeanFactory(applicationContext.getAutowireCapableBeanFactory());
        postProcessor.afterPropertiesSet();
        return postProcessor;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }


    @Autowired
    private JobLauncher jobLauncher;
    @Autowired
    private JobExplorer jobExplorer;

    @Autowired
    private JobRepository jobRepository;

    @Autowired
    private JobRegistry jobRegistry;
    @Autowired
    MyItemWriter myItemWriter;

    @Bean
    public JobOperator jobOperator() {
        SimpleJobOperator jobOperator = new SimpleJobOperator();
        jobOperator.setJobLauncher(jobLauncher);
        jobOperator.setJobRepository(jobRepository);
        jobOperator.setJobExplorer(jobExplorer);
        jobOperator.setJobRegistry(jobRegistry);
        jobOperator.setJobParametersConverter(new DefaultJobParametersConverter());

        return jobOperator;
    }

    @Bean
    public FlatFileItemReader<User> itemReader()  {
        FlatFileItemReader<User> fileTxtFlatFileItemReader =
                new FlatFileItemReaderBuilder<User>().name("fileTxtIn")
                        .resource(new ClassPathResource("file/user.txt")).delimited().delimiter("#")
                        .names("id", "name", "age").targetType(User.class).build();

        return fileTxtFlatFileItemReader;
    }

    @Bean
    public ItemWriter<User> itemWriter() {
        FlatFileItemWriter<User> writer =
                new FlatFileItemWriterBuilder<User>().name("fileTxtOut")
                        .resource(new FileSystemResource("e:/userOutNew.txt")).delimited().delimiter("#")
                        .names("id", "name", "age").build();

        return writer;
    }

    @Autowired
    @Qualifier("itemWriteDb")
    ItemWriter<? super User>  itemWriteDb;
    @Bean
    public JdbcBatchItemWriter itemWriteDb() throws Exception {
        log.info("itemWriter2Table {} = ", batchNo.addAndGet(1));

        JdbcBatchItemWriter writer = new JdbcBatchItemWriter();
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider());
        writer.setSql("insert into person(id,name,age) values(:id,:name,:age)");// would like to insert into pharmacy table);
        writer.setDataSource(pfDatasource);
        // claimWriter.afterPropertiesSet();
        return writer;

    }


    @Bean
    UserRowMapper userRowMapper(){
        return new UserRowMapper();

    }
    @Bean
    public CompositeItemWriter<User> compositeItemWriter() {
        CompositeItemWriter writer = new CompositeItemWriter();
        writer.setDelegates(Arrays.asList(myItemWriter,itemWriteDb));
        return writer;
    }


    @Component("myItemWriter")
    public class MyItemWriter implements ItemWriter<User>, Serializable {
        @Override
        public void write(List<? extends User> list) throws Exception {
            log.info("输出batchNo =  {}, ", batchNo.addAndGet(1));
            for (User user : list) {
                log.info("item = {}",JSONObject.toJSONString(user));
            }
        }
    }
    @Bean
    public PagingQueryProvider pagingQueryProvider() throws Exception {
        SqlPagingQueryProviderFactoryBean factoryBean = new SqlPagingQueryProviderFactoryBean();
        factoryBean.setDataSource(pfDatasource);
        factoryBean.setSelectClause("select *");   //查询列
        factoryBean.setFromClause("from user");    //查询的表
        factoryBean.setWhereClause("where age > :age"); //where 条件
        factoryBean.setSortKey("id");   //结果排序
        return factoryBean.getObject();
    }
    @Bean
    public JdbcPagingItemReader<User> userItemReader1() throws Exception {
        HashMap<String, Object> param = new HashMap<>();
        param.put("age", 1);
        return new JdbcPagingItemReaderBuilder<User>()
                .name("userPagingItemReader")
                .dataSource(pfDatasource)  //数据源
                .queryProvider(pagingQueryProvider())  //分页逻辑
                .parameterValues(param)   //条件
                .pageSize(10) //每页显示条数
                .rowMapper(userRowMapper())  //映射规则
                .build();
    }
    @Bean
    public JdbcCursorItemReader<User> userItemReader(){

        return new JdbcCursorItemReaderBuilder<User>()
                .name("userCursorItemReader")
                .dataSource(pfDatasource)
                .sql("select * from person limit 8 ")
                .rowMapper(userRowMapper())
                .build();
    }

    @Bean
    public Step simpleStep0() throws Exception {
        return stepBuilderFactory.get("simpleStep0")
                .<User, User>chunk(1).reader(itemReader())
                // .writer(itemWriter())
                .writer(myItemWriter)
                .build();
    }
    @Bean
    public Step simpleStepReadDb2File()  {
        return stepBuilderFactory.get("simpleStepReadDb2File")
                .<User, User>chunk(4).reader(userItemReader())
                .writer(compositeItemWriter()) //compositeItemWriter
                .build();
    }

    @Bean
    public Step simpleStep1() throws Exception {
        return stepBuilderFactory.get("simpleStep1")
                .<User, User>chunk(1).reader(itemReader())
                .writer(itemWriter())
                .build();
    }

    @Bean
    public Step simpleStep2Table() throws Exception {
        return stepBuilderFactory.get("simpleStep2Table")
                .<User, User>chunk(2).reader(itemReader())
                .writer(itemWriteDb())
                .build();
    }

    public Step simpleStep2() {
        return stepBuilderFactory.get("simpleStep2")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("Hello, World!");
                    //List list = metadataModel.selectMysqlColumns("employee");
                    //log.info("list={}", CommonUtils.toJSONString(list));
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    @Bean
    public Job simpleJobNew() throws Exception {
        return jobBuilderFactory.get("simpleJobNew")
                .start(simpleStep1())
                .build();
    }

    @Bean
    public Job simpleJob0() throws Exception {
        return jobBuilderFactory.get("simpleJob0")
                .start(simpleStep0())
                .build();
    }

    @Bean
    public Job simpleJob2Table() throws Exception {
        return jobBuilderFactory.get("simpleJob2Table")
                .start(simpleStep2Table())
                .build();
    }
    @Bean
    public Job simpleJobReadDb2File() throws Exception {
        return jobBuilderFactory.get("simpleJobReadDb2File")
                .start(simpleStepReadDb2File())
                .build();
    }



}